package com.zxinlee.m3;

import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.common.message.MessageExt;

public class Consumer {

    public static void main(String[] args) throws MQClientException {
        //创建消费者
        DefaultMQPushConsumer consumer01 = new DefaultMQPushConsumer("consumer03");

        //设置 name server
        consumer01.setNamesrvAddr("192.168.64.141:9876");

        //设置从哪订阅消息
        consumer01.subscribe("Topic03", "*");

        //处理消息的监听器
        //Concurrently 会启动多个线程,并发处理消息
        consumer01.setMessageListener((MessageListenerConcurrently) (msgs, context) -> {
            for (MessageExt msg : msgs) {
                System.out.println("收到消息: " + new String(msg.getBody()));
            }
            //return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            System.out.println("处理消息失败");
            return ConsumeConcurrentlyStatus.RECONSUME_LATER;
        });

        //启动消费者
        consumer01.start();

    }
}
